package com.huatai.datacenter.service;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.huatai.datacenter.constant.Constants;
import com.huatai.datacenter.entity.BrokerEntity;
import com.huatai.datacenter.entity.postdata.HttpRequestEntity;
import com.huatai.datacenter.mapper.ClusterMapper;
import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.utils.KafkaAdmins;
import kafka.zk.BrokerInfo;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.*;

/**
 * @author Lion
 * @date 2023/3/20  10:49
 */
@Service
public class ClusterService extends BaseService<ClusterMapper, ClusterEntity> {

	@Autowired
	private ClusterMapper clusterMapper;

	@Autowired
	private ZKService zkService;
	@Autowired
	private KafkaAdminService kafkaAdminService;

	@Autowired
	private TopicInfoService topicInfoService;

	/**
	 * @param cluster
	 * @return boolean
	 * @author Lion
	 * @description 新增kafka服务端
	 */
	public boolean insert(ClusterEntity cluster) {
		Integer result = 0;
		try {
			if (zkService.checkServiceHealth(cluster.getZkpAddress()) && kafkaAdminService.kafkaIsHeath(cluster.getBroker())) {
				cluster.setCreateTime(LocalDateTime.now());
				result = clusterMapper.insert(cluster);
				return checkResult(result);
			}
		} catch (Exception e) {
			return false;
		}
		return false;
	}

	/**
	 * @param clusterId
	 * @return java.lang.String
	 * @author Lion
	 * @description 根据客户端id获取所有kafka集群信息
	 */
	public String getBrokers(String clusterId) {
		ClusterEntity clusterEntity = clusterMapper.selectById(clusterId);
		return clusterEntity.getBroker();
	}

	/**
	 * @Description: 根据broker获取详情
	 * @Param: clusterId
	 * @return: java.lang.String
	 * @Author: Yukun
	 * @Date: 2025/5/7
	 */
	public String getBrokersByClusterId(String clusterId) {
		QueryWrapper<ClusterEntity> queryWrapper = new QueryWrapper<>();
		queryWrapper.eq("broker",clusterId);
		ClusterEntity clusterEntity = clusterMapper.selectOne(queryWrapper);
		return clusterEntity.getBroker();
	}

	/**
	 * @param cluster
	 * @param isUpdate
	 * @return boolean
	 * @author Lion
	 * @description 校验地址
	 */
	public boolean clusterExits(ClusterEntity cluster, boolean isUpdate) {
		try {
			List<ClusterEntity> clusterEntities = this.list();
			//如果是修改，先把自身移除
			if (isUpdate) {
				clusterEntities.removeIf(clusterInfo -> clusterInfo.getId().equals(cluster.getId()));
			}
			// 先校验name
			for (ClusterEntity clusterInfo : clusterEntities) {
				if (clusterInfo.getClusterName().equals(cluster.getClusterName())) {
					return true;
				}
			}
			//再校验zk/broker
			Set<String> brosSet = new HashSet<>(Arrays.asList(cluster.getBroker().split(Constants.Symbol.COMMA)));
			for (ClusterEntity clusterInfo : clusterEntities) {
				List<BrokerEntity> brokerEntities = zkService.getZK(clusterInfo.getId().toString()).getBrokers();
				Set<String> brokers = new HashSet<>();
				brokerEntities.forEach(broker -> {
					String brk = broker.getHost() + Constants.Symbol.COLON + broker.getPort();
					brokers.add(brk);
				});
				Set<String> result = intersection(brokers, brosSet);
				if (!result.isEmpty()) {
					return true;
				}
			}
		} catch (Exception e) {
			return false;
		}
		return false;
	}

	public Set<String> intersection(Set<String> s1, Set<String> s2) {
		Set<String> result = new HashSet<>();
		result.clear();
		result.addAll(s1);
		result.retainAll(s2);
		return result;
	}

	private boolean checkResult(Integer result) {
		return result > 0;
	}

	public ClusterEntity selectById(Long id) {
		return clusterMapper.selectById(id);
	}

	public List<ClusterEntity> getTotalData() {
		return clusterMapper.selectList(null);
	}

	/**
	 * @author Lion
	 * @description 删除kafka集群
	 * @param  * @param id
	 * @return boolean
	 */
	/*public boolean deleteAssociateTable(Long id) {
		boolean topic = topicInfoService.deleteByClusterId(id);
		boolean alter = alterService.deleteByClusterId(id);
		boolean task = taskInfoService.deleteByClusterId(id);
		boolean collect;
		Map<String,Object> map = new HashMap<>();
		map.put(Constants.KeyStr.CLUSTER_ID, id);
		if(collectionService.selectByMap(map)) {
			collect = collectionService.deleteByMap(map);
		}else {
			collect = true;
		}
		return topic && alter && task && collect;
	}*/

	/**
	 * @param cluster
	 * @return com.alibaba.fastjson.JSONObject
	 * @author Lion
	 * @description 判断kafka的状态
	 */
	public void getClusterAndStatus(ClusterEntity cluster) {

		//kafka健康检查
		if (kafkaAdminService.kafkaIsHeath(cluster.getBroker())) {
			cluster.setStatus(1);
		} else {
			cluster.setStatus(0);
		}
		this.updateById(cluster);
	}
}
